

package com.coding4play.mqtt.store.impl;

import cn.hutool.core.util.StrUtil;
import com.coding4play.mqtt.common.entity.message.RetainMessage;
import com.coding4play.mqtt.common.repository.IRetainMessageRepository;
import com.coding4play.mqtt.store.config.PropertiesConfig;
import com.coding4play.mqtt.store.config.RedisTemplateWrapper;
import com.coding4play.mqtt.store.util.RedisScan;
import com.coding4play.mqtt.store.util.StringTopicMatch;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.stream.Collectors;

@Component
public class RetainMessageRepositoryImpl extends RedisTemplateWrapper implements IRetainMessageRepository {

    private final static String SET_RETAIN_PRE = "mqtt:retain:topic_msg:";

    private PropertiesConfig propertiesConfig;

    public RetainMessageRepositoryImpl(RedisTemplate<String, Object> redisTemplate,
                                       PropertiesConfig propertiesConfig) {
        super(redisTemplate);
        this.propertiesConfig = propertiesConfig;
    }

    @Override
    public void put(String topic, RetainMessage retainMessage) {
        KV.set(getKey(topic), retainMessage);
    }

    @Override
    public RetainMessage get(String topic) {
        Object obj = KV.get(getKey(topic));
        return (RetainMessage) obj;
    }

    @Override
    public void delete(String topic) {
        redisTemplate.delete(getKey(topic));
    }

    @Override
    public Boolean containsKey(String topic) {
        return redisTemplate.hasKey(getKey(topic));
    }

    @Override
    public List<RetainMessage> search(String topicFilter) {
        if (!StrUtil.contains(topicFilter, '#') && !StrUtil.contains(topicFilter, '+')) {
            RetainMessage retainMessage = get(topicFilter);
            return Objects.isNull(retainMessage) ? Collections.emptyList() : Collections.singletonList(retainMessage);
        }

        List<String> keyList = RedisScan.keyScan(
                redisTemplate.getConnectionFactory(), propertiesConfig.isRedisCluster, getKey("*")
        );

        int length = SET_RETAIN_PRE.length();
        List<RetainMessage> collect = keyList.stream()
                .filter(k -> StringTopicMatch.match(k.substring(length), topicFilter))
                .map(k -> KV.get(k))
                .map(obj -> (RetainMessage) obj)
                .collect(Collectors.toList());
        return collect;
    }


    private String getKey(String topic) {
        return SET_RETAIN_PRE + topic;
    }
}
